package com.ku.order.service.mq;

import cn.hutool.json.JSONUtil;
import com.ku.constants.OrderStatus;
import com.ku.entity.Order;
import com.ku.order.mapper.OrderMapper;
import com.ku.service.OrderService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;

/**
 * @program: rocketMQKill
 * @description: 用户支付后，回调回来，需要修改订单的状态， 由下单改为已支付
 * @author: JokerKu
 * @create: 2020-05-07 10:54
 **/
@Component
public class SyncConsumer {

    private DefaultMQPushConsumer consumer;

    public static final String CALLBACK_PAY_TOPIC = "callback_pay_topic";

    public static final String CALLBACK_PAY_TAGS = "callback_pay";

    private static final String NAME_SERVER = "192.168.43.10:9876;192.168.43.9:9876";

    private static final String CONSUMER_GROUP_NAME = "callback_pay_consumer_group";

    @Autowired
    private OrderMapper orderMapper;
    @Autowired
    private OrderService orderService;

    private SyncConsumer() {
        try {
            this.consumer = new DefaultMQPushConsumer(CONSUMER_GROUP_NAME);
            this.consumer.setNamesrvAddr(NAME_SERVER);
            this.consumer.setConsumeThreadMin(10);
            this.consumer.setConsumeThreadMax(30);
            this.consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            this.consumer.subscribe(CALLBACK_PAY_TOPIC, CALLBACK_PAY_TAGS);
            this.consumer.registerMessageListener(new MessageListenerConcurrentlyImpl());
            this.consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }

    }

    class MessageListenerConcurrentlyImpl implements MessageListenerConcurrently {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgList, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
            MessageExt messageExt = msgList.get(0);
            try {
                String topic = messageExt.getTopic();
                String keys = messageExt.getKeys();
                String tags = messageExt.getTags();
                System.err.println("topic:" + topic + "---keys:" + keys + "---tags:" + tags);
                //获取消息
                String msg = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
                //转换为Map
                Map<String, Object> params = JSONUtil.toBean(msg, Map.class);
                String orderId = (String) params.get("orderId");
                String userId = (String) params.get("userId");
                String oderStatus = (String) params.get("OderStatus");
                //调用orderMapper修改订单状态
                Order order = new Order();
                order.setOrderId(orderId);
                order.setUserId(userId);
                order.setOrderStatus(oderStatus);
                int updateResult = orderMapper.updateByPrimaryKeySelective(order);
                if (updateResult==1){
                    //修改订单状态成功 这时需要创建包裹、发送物流通知
                    orderService.sendOrderlyMessageForPkg(userId,orderId);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            } catch (Exception e) {
                e.printStackTrace();
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
}
